package com.xahj.bd2104.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created with IntelliJ IDEA.
 * Author: Amos
 * E-mail: amos@amoscloud.com
 * Date: 2021/9/13
 * Time: 10:39
 * Description: 
 */
object SparkCoreDemo5_transformations {
  def main(args: Array[String]): Unit = {
    // RDD的转换算子
    val conf = new SparkConf()
      .setAppName("Demo5")
      .setMaster("local[2]")
    val sc = new SparkContext(conf)

    val rdd: RDD[Int] = sc.makeRDD(1 to 10)

    // 1. map
    //  通过传入一元函数，对源RDD中的元素进行逐一处理
    //  map可以改变元素的类型
    // def map[U: ClassTag](f: T => U): RDD[U] =
    //                     Int=>String
    val rdd1: RDD[String] = rdd.map(x => (x + 1) + "")

    // 2. filter
    //def filter(f: T => Boolean): RDD[T] =
    val rdd2 = rdd.filter(x => x % 2 != 0)

    // 3. flatMap
    //                                   👇集合的超类
    // def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U] =
    // flatMap接受的参数f 的返回值必须时一个集合类型
    val rdd3 = sc.makeRDD(List("asadsf", "asdasd", "rethyt"))
    val rdd4: RDD[Char] = rdd3.flatMap(x => x)

    // 4. mapPartitions
    //  传入一个一元函数f
    //   f每次拿到一整个分区的数据，处理完返回整个分区的数据
    //  def mapPartitions[U: ClassTag](
    //      f: Iterator[T] => Iterator[U],
    //      preservesPartitioning: Boolean = false): RDD[U] =
    val rdd5 = sc.makeRDD(1 to 100)

    val rdd6: RDD[(Int, Int, Int)] = rdd5.mapPartitions(iter => {
      val arr = iter.toArray
      val sum = arr.sum
      val size = arr.length
      val avg = sum / size
      Iterator((sum, size, avg))
    })

    // 5. mapPartitionsWithIndex
    //  def mapPartitionsWithIndex[U: ClassTag](
    //      f: (Int, Iterator[T]) => Iterator[U],
    //      preservesPartitioning: Boolean = false): RDD[U] =
    rdd5.mapPartitionsWithIndex((index, iter) => {
      val str = iter.mkString(",")
      Iterator(index + "-" + str)
    })
    //      .foreach(println)

    // 6. sample
    //  def sample(
    //      // 设置为true ，相同的元素可以被抽取多次
    //      withReplacement: Boolean,
    //      非精确抽样比例，因为RDD的每个分区都会启动单独的线性进行操作
    //      而多个线程之间不进行数量的互通
    //      每个线程都根据比例在自己分区内进行抽样
    //      随着样本(RDD元素)数量增加，抽样比例会趋近于精确
    //      fraction: Double,   0~1
    //      随机种子  用于抽样算法的起始值
    //      相同种子可以拿到相同抽样结果
    //      seed: Long = Utils.random.nextLong): RDD[T] =

    // 通常用于海量数据  抽取样本进行测试
    val rdd7 = rdd5.sample(false, 0.1, 17)

    // 7. union
    // 合并两个泛型相同的RDD   不去重
    //def union(other: RDD[T]): RDD[T] =
    val rdd8 = sc.makeRDD(1 to 5)
    val rdd9 = sc.makeRDD(3 to 8)

    val rdd10 = rdd8.union(rdd9)
    val arr = rdd10.collect()

    // 8. intersection    ∩
    // 两个相同泛型RDD的交集
    val rdd11 = rdd8.intersection(rdd9)
    //    rdd11.foreach(println)

    // 9. distinct
    // distinct(新rdd的分区数)
    // 通常由于去重会缩减元素个数
    // 所以可以适当缩减分区个数
    // 减小并发度
    val rdd12 = sc.makeRDD(List(1, 1, 1, 1, 2, 3, 4, 4, 3))
    val rdd13: RDD[Int] = rdd12.distinct()
    rdd13.foreach(println)


    val rdd14: RDD[(Int, Int)] = rdd13.map((_, 1))
  }
}
